package com.shujia.stream


import java.util.{Date, Properties}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.io.Source

object Demo6WiterKafka {
  def main(args: Array[String]): Unit = {

    /**
      * 模拟数据生成
      *
      */

    //定义配置文件对象
    val properties = new Properties
    properties.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092") //kafka连接地址
    properties.put("key.serializer", classOf[StringSerializer])
    properties.put("value.serializer", classOf[StringSerializer])

    //创建连接   创建生产者
    val producer = new KafkaProducer[String, String](properties)

    Source
      .fromFile("data/dianxin")
      .getLines()
      .foreach(line => {
        //以时间戳作为偏移量
        val key = new Date().getTime.toString
        //构建消息对象
        val record = new ProducerRecord[String, String]("topic4", key, line)

        //发送消息
        producer.send(record)
        println("发送数据：" + line)

        Thread.sleep(1000)
      })

  }
}
